package com.zhang.flink.example;

import com.zhang.flink.bean.LoginEvent;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;


/**
 * @title: 恶意登录检测(使用KeyedProcessFunction实现）
 * 存在缺陷如果业务逻辑需要连续两秒失败3次、4次 就需要改动代码
 * @author: zhang
 * @date: 2022/2/13 00:05
 */
public class MaliceLoginDetect {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .readTextFile("src/main/resources/LoginLog.csv")
                .map(new MapFunction<String, LoginEvent>() {
                    @Override
                    public LoginEvent map(String value) throws Exception {
                        String[] fields = value.split(",");
                        return new LoginEvent(
                                fields[0], fields[1], fields[2],
                                Long.parseLong(fields[3]) * 1000L
                        );
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<LoginEvent>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<LoginEvent>() {
                                    @Override
                                    public long extractTimestamp(LoginEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                )
                .keyBy(r -> r.userId)
                .process(new FailLoginProcess(2))
                .print();


        env.execute();
    }

    public static class FailLoginProcess extends KeyedProcessFunction<String, LoginEvent, String> {
        private long maxTime;
        private ValueState<LoginEvent> loginEventState;

        public FailLoginProcess(long maxTime) {
            this.maxTime = maxTime * 1000L;
        }

        @Override
        public void open(Configuration parameters) throws Exception {
            loginEventState = getRuntimeContext().getState(
                    new ValueStateDescriptor<LoginEvent>(
                            "login-state", Types.POJO(LoginEvent.class)
                    )
            );
        }

        @Override
        public void processElement(LoginEvent value, KeyedProcessFunction<String, LoginEvent, String>.Context ctx, Collector<String> out) throws Exception {
            if ("fail".equals(value.type)) {

                LoginEvent loginEvent = loginEventState.value();
                loginEventState.update(value);

                //如果不是一次失败，取出状态，判断时间是不是在两秒内
                if (loginEvent != null && Math.abs(value.ts - loginEvent.ts) <= maxTime) {
                    out.collect(value.userId + "连续登录失败2次，报警！！！");
                }
            } else {
                loginEventState.clear();
            }
        }
    }

}
